package mapreduce;
import java.io.IOException;
import java.util.Iterator;
import java.util.Vector;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
public class ReduceJoin {
    public static class mymapper extends Mapper < Object, Text, Text, Text > {
        @Override
        protected void map(Object key, Text value, Context context)
        throws IOException,
        InterruptedException {
            String filePath = ((FileSplit) context.getInputSplit()).getPath().toString();
            if (filePath.contains("orders1")) {
                String line = value.toString();
                String[] arr = line.split("\t");
                context.write(new Text(arr[0]), new Text("1+" + arr[2] + "\t" + arr[3]));
                //System.out.println(arr[0] + "_1+" + arr[2]+"\t"+arr[3]);  
            } else if (filePath.contains("order_items1")) {
                String line = value.toString();
                String[] arr = line.split("\t");
                context.write(new Text(arr[1]), new Text("2+" + arr[2]));
                //System.out.println(arr[1] + "_2+" + arr[2]);  
            }
        }
    }

    public static class myreducer extends Reducer < Text, Text, Text, Text > {
        @Override
        protected void reduce(Text key, Iterable < Text > values, Context context)
        throws IOException,
        InterruptedException {
            Vector < String > left = new Vector < String > ();
            Vector < String > right = new Vector < String > ();
            for (Text val: values) {
                String str = val.toString();
                if (str.startsWith("1+")) {
                    left.add(str.substring(2));
                } else if (str.startsWith("2+")) {
                    right.add(str.substring(2));
                }
            }

            int sizeL = left.size();
            int sizeR = right.size();
            //System.out.println(key + "left:"+left);  
            //System.out.println(key + "right:"+right);  
            for (int i = 0; i < sizeL; i++) {
                for (int j = 0; j < sizeR; j++) {
                    context.write(key, new Text(left.get(i) + "\t" + right.get(j)));
                    //System.out.println(key + " \t" + left.get(i) + "\t" + right.get(j));  
                }
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Job job = Job.getInstance();
        job.setJobName("reducejoin");
        job.setJarByClass(ReduceJoin.class);

        job.setMapperClass(mymapper.class);
        job.setReducerClass(myreducer.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        Path left = new Path("hdfs://localhost:9000/mymapreduce6/in/orders1");
        Path right = new Path("hdfs://localhost:9000/mymapreduce6/in/order_items1");
        Path out = new Path("hdfs://localhost:9000/mymapreduce6/out");

        FileInputFormat.addInputPath(job, left);
        FileInputFormat.addInputPath(job, right);
        FileOutputFormat.setOutputPath(job, out);

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}